package com.atguigu.dwm.model.service.impl;

import com.atguigu.dwm.common.component.TableHdfsProcessor;
import com.atguigu.dwm.common.component.TableHiveProcessor;
import com.atguigu.dwm.common.constants.CommonCodes;
import com.atguigu.dwm.common.mapper.HiveJdbcMapper;
import com.atguigu.dwm.model.bean.DmTable;
import com.atguigu.dwm.model.bean.DmTableDataInfo;
import com.atguigu.dwm.model.bean.DmTableSync;
import com.atguigu.dwm.model.mapper.DmTableMapper;
import com.atguigu.dwm.model.mapper.DmTableSyncMapper;
import com.atguigu.dwm.model.service.DmTableDataInfoService;
import com.atguigu.dwm.model.service.DmTableService;
import com.atguigu.dwm.model.service.DmTableSyncService;
import com.atguigu.dwm.plan.bean.DpDataWarehouseModel;
import com.atguigu.dwm.plan.service.DpDataWarehouseModelService;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.thrift.TException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import com.baomidou.dynamic.datasource.annotation.DS;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 数据同步信息表 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2024-07-05
 */
@Service
@DS("dwm")
public class DmTableSyncServiceImpl extends ServiceImpl<DmTableSyncMapper, DmTableSync> implements DmTableSyncService {

    @Autowired
    TableHiveProcessor tableHiveProcessor;

    @Autowired
    DmTableService dmTableService;

    @Autowired
    DmTableMapper dmTableMapper;

    @Autowired
    DpDataWarehouseModelService dpDataWarehouseModelService;

    //1  要根据库名查询hive的数据库表
    //2  再根据表名查询已保存过的同步信息（table_sync ）
    //3  如果没有查到同步信息  ,查询是否有同名的建模表
    //4  如果建模表也没有，查一下有多同名schema的数仓模型
    //5  把以上组合为一个同步信息列表List<DmTableSync>返回给页面
    @Override
    public List<DmTableSync> getSyncList(String schemaName) throws Exception {
        //1  要根据库名查询hive的数据库表
        List<String> tableNameList = tableHiveProcessor.getTableNameList(schemaName);

        //2  准备已经保存的同步信息列表  为了方便匹配 转为map  key : tableName  value : DmTableSync
        List<DmTableSync> dmTableSyncList = this.list(new QueryWrapper<DmTableSync>().eq("schema_name", schemaName));
        Map<String, DmTableSync> tableSyncMap = dmTableSyncList.stream().collect(Collectors.toMap(DmTableSync::getTableName, dmTableSync -> dmTableSync));
        // 等效： dmTableSyncList.stream().collect(Collectors.toMap(dmTableSync -> dmTableSync.getTableName(), dmTableSync -> dmTableSync));

        //3 准备建模表的列表  为了方便匹配 转为map  key : tableName  value : DmTable
        QueryWrapper<DmTable> queryWrapper = new QueryWrapper<DmTable>().eq("schema_name", schemaName)
                .eq("is_deleted", "0");
        List<DmTable> dmTableList = dmTableService.list(queryWrapper);
        Map<String, DmTable> dmTableMap = dmTableList.stream().collect(Collectors.toMap(DmTable::getTableName, dmTable -> dmTable));

        //4 准备数仓模型的列表  为了方便匹配 转为map  key : schema  value : DpDataWarehousrModel
        List<DpDataWarehouseModel> modelList = dpDataWarehouseModelService.list(new QueryWrapper<DpDataWarehouseModel>().eq("is_deleted", "0"));
        Map<String, DpDataWarehouseModel> modelMap = modelList.stream().collect(Collectors.toMap(DpDataWarehouseModel::getSchemaName, dpDataWarehouseModel -> dpDataWarehouseModel));

        //5 循环 hive列表
        //5 1  再根据表名查询已保存过的同步信息（table_sync ）
        //5 2  如果没有查到同步信息  ,查询是否有同名的建模表
        //5 3   如果建模表也没有，查一下有多同名schema的数仓模型
        //5 4 把以上组合为一个同步信息列表List<DmTableSync>返回给页面
        List<DmTableSync> dmTableSyncListForShow = new ArrayList<>();
        for (String tableName : tableNameList) {
            DmTableSync dmTableSync = new DmTableSync();

            if (tableSyncMap.containsKey(tableName)) {
                //5 1  再根据表名查询已保存过的同步信息（table_sync ）
                dmTableSync = tableSyncMap.get(tableName);
            } else {
                //5 2  如果没有查到同步信息  ,查询是否有同名的建模表
                if (dmTableMap.containsKey(tableName)) {
                    //5 2 1  如果有同名的建模表，把建模表的信息赋值给同步信息
                    DmTable dmTable = dmTableMap.get(tableName);
                    dmTableSync.setModelId(dmTable.getModelId());
                    dmTableSync.setSchemaName(dmTable.getSchemaName());
                    dmTableSync.setTableId(dmTable.getId());
                    dmTableSync.setTableName(dmTable.getTableName());


                } else {
                    //5 3   如果建模表也没有，查一下有多同名schema的数仓模型
                    DpDataWarehouseModel model = modelMap.get(schemaName);
                    if (model != null) {
                        dmTableSync.setModelId(model.getId());

                    }
                }
            }
            dmTableSync.setSchemaName(schemaName);
            dmTableSync.setTableName(tableName);
            dmTableSyncListForShow.add(dmTableSync);
        }
//5 4 把以上组合为一个同步信息列表List<DmTableSync>返回给页面

        return dmTableSyncListForShow;
    }


    // 1   问题： 同步表数据 用不用查询出原表的子表（字段信息)
    //  2  把hive中的对提取出来  装配给dmTable建表
    //
    @Override
    public void syncMeta(List<DmTableSync> tableSyncList) throws TException {

        List<Long> tableIdList = tableSyncList.stream().filter(dmTableSync -> dmTableSync.getTableId() != null)
                .map(dmTableSync -> dmTableSync.getTableId()).collect(Collectors.toList());
        Map<Long, DmTable> dmTableMap = new HashMap<>();
        if (tableIdList.size() > 0) {
            List<DmTable> dmTableList = dmTableService.list(new QueryWrapper<DmTable>().in("id", tableIdList));
            //把dmTableList转为map 用于tableId作为key 方便后续使用
            dmTableMap = dmTableList.stream().collect(Collectors.toMap(DmTable::getId, dmTable -> dmTable));

        }


        for (DmTableSync dmTableSync : tableSyncList) {
            DmTable dmTable = null;
            if (dmTableSync.getTableId() != null) {
                dmTable = dmTableMap.get(dmTableSync.getTableId());
            } else {
                dmTable = new DmTable();
                dmTable.setSchemaName(dmTableSync.getSchemaName());
                dmTable.setTableName(dmTableSync.getTableName());
                dmTable.setModelId(dmTableSync.getModelId());
            }

            //2  把hive中的对提取出来  装配给dmTable建表
            tableHiveProcessor.syncTableMeta(dmTable);

            //3 保存dmtable建模表
            dmTable.setTableStatus(CommonCodes.TABLE_STATUS_ADDED);
            dmTableService.saveTableAll(dmTable);

            //4 保存 对应的同步信息
            dmTableSync.setTableId(dmTable.getId());
            dmTableSync.setLastSyncMetaTime(new Date());
            dmTableSync.setLastSyncMetaUserId(9999L);
            this.saveOrUpdate(dmTableSync);
        }


    }

    @Autowired
    DmTableDataInfoService dmTableDataInfoService;

    @Autowired
    HiveJdbcMapper hiveJdbcMapper;

    @Autowired
    TableHdfsProcessor tableHdfsProcessor;


    @Override
    public void syncDataInfo(List<DmTableSync> tableSyncList) throws Exception {
        for (DmTableSync dmTableSync : tableSyncList) {
            //强制统计数据信息
         //   hiveJdbcMapper.analyzeTable(dmTableSync.getSchemaName(), dmTableSync.getTableName());
            DmTableDataInfo dmTableDataInfo = new DmTableDataInfo();
            //提前统计信息
            List<Map<String, Object>> tableFormattedInfoList= hiveJdbcMapper.getTableFormattedInfo(dmTableSync.getSchemaName(), dmTableSync.getTableName());

             for (Map<String, Object> formatMap : tableFormattedInfoList) {
                 if (formatMap.get("data_type")==null){
                     continue;
                 }
                 String dataType = formatMap.get("data_type").toString().trim();
                 if(dataType.equals("numFiles")){ //文件大小
                    String numFiles = formatMap.get("comment").toString().trim();
                    dmTableDataInfo.setNumFiles(Long.valueOf(numFiles));
                }else  if(dataType.equals("numPartitions")){//分区个数
                    String numPartitions = formatMap.get("comment").toString().trim();
                    dmTableDataInfo.setNumPartitions(Long.valueOf(numPartitions));
                }else  if(dataType.equals("rawDataSize")){//未压缩的数据(估算)
                    String rawDataSize = formatMap.get("comment").toString().trim();
                    dmTableDataInfo.setRawDataSize(Long.valueOf(rawDataSize));
                }else  if(dataType.equals("numRows")){//行数
                    String numRows = formatMap.get("comment").toString().trim();
                    dmTableDataInfo.setNumRows(Long.valueOf(numRows));
                }else  if(dataType.equals("totalSize")){//数据大小单副本
                    String totalSize = formatMap.get("comment").toString().trim();
                    dmTableDataInfo.setDataSize(Long.valueOf(totalSize));
                }
            }
             if(dmTableDataInfo.getRawDataSize()!=null&&dmTableDataInfo.getRawDataSize()!=0L){
                 //计算 压缩比
                 dmTableDataInfo.setCompressRatio(   BigDecimal.valueOf(dmTableDataInfo.getDataSize()).divide(BigDecimal.valueOf(dmTableDataInfo.getRawDataSize()), 2, BigDecimal.ROUND_HALF_UP)   );
             }
            if(dmTableDataInfo.getNumFiles()!=null&&dmTableDataInfo.getNumFiles()!=0L) {
                //计算 文件平均大小
                dmTableDataInfo.setFileSizeAvg(BigDecimal.valueOf(dmTableDataInfo.getDataSize()).divide(BigDecimal.valueOf(dmTableDataInfo.getNumFiles()), 2, BigDecimal.ROUND_HALF_UP).longValue());
            }
            dmTableDataInfo.setTableId(dmTableSync.getTableId());
            dmTableDataInfo.setTableName(dmTableSync.getTableName());
            dmTableDataInfo.setSchemaName(dmTableSync.getSchemaName());
            dmTableDataInfo.setLastSyncInfoTime(new Date());


            //获得hdfs的相关信息： 副本大小 最后访问时间 最后修改时间
            Table table = tableHiveProcessor.getTable(dmTableSync.getSchemaName(), dmTableSync.getTableName());
            tableHdfsProcessor.getHdfsFileInfo(table,dmTableDataInfo);

            dmTableDataInfoService.remove(new QueryWrapper<DmTableDataInfo>()
                    .eq("table_name",dmTableDataInfo.getTableName())
                    .eq("schema_name",dmTableDataInfo.getSchemaName()));

            dmTableDataInfoService.saveOrUpdate(dmTableDataInfo);
            dmTableSync.setLastSyncInfoTime(new Date());
            this.saveOrUpdate(dmTableSync);
        }
    }


    public void  syncDataInfoForScheduler() throws Exception {
        //1  查询出需要同步信息的列表
        List<DmTableSync> dmTableSyncList = this.list(new QueryWrapper<DmTableSync>().eq("is_sync_info", "1"));

        // 2遍历该列表 逐个同步信息
        syncDataInfo(  dmTableSyncList);


    }
}
